package com.gzzz.bean.impl;

import com.gzzz.bean.Input;
import com.gzzz.config.KafkaConfig;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * @author leiming
 */

public class KafkaInput implements Input {

    public final static String TOPIC_NAME = "CustomTopic";
    public KafkaConsumer<String, String> consumer;

    public KafkaInput() {
        System.out.println("Get stream from Kafka");
        Properties properties = new Properties();
        properties.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, KafkaConfig.BOOTSTRAP_SERVERS_CONFIG);
        properties.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        properties.put(ConsumerConfig.GROUP_ID_CONFIG, KafkaConfig.GROUP_ID_CONFIG);
        properties.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, KafkaConfig.AUTO_OFFSET_RESET_CONFIG);
        properties.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, KafkaConfig.ENABLE_AUTO_COMMIT_CONFIG);
        consumer = new KafkaConsumer<>(properties);
    }

    /**
     * 从 Kafka 消息队列中读取单个消息，并以输入流的方式返回
     * 此处暂未考虑多个消息的处理
     * @return InputStream
     */
    public InputStream read() {
        consumer.subscribe(Arrays.asList(TOPIC_NAME));
        ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(5));
        // consume only one message for now
        for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
            String value = consumerRecord.value();
            System.out.println("Get Value :" + value);
            byte[] byteArray = value.getBytes(StandardCharsets.UTF_8);
            return new ByteArrayInputStream(byteArray);
        }
        return null;
    }
}
